﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using log4net;
using Newtonsoft.Json;
using TdxHq.Common;
using TdxHq.Common.Extend;
using TdxHq.Connect;
using TdxHq.ErrorHandler;
using TdxHq.Invoker;
using TdxHq.Looper;

namespace TdxHq
{
    class SanpshootService
    {
        private readonly ILog _log = LogManager.GetLogger(typeof(SanpshootService));

        public static readonly ConcurrentDictionary<string, QuoteRt> Snapshoot =
            new ConcurrentDictionary<string, QuoteRt>();
        public static readonly ConcurrentQueue<QuoteRt> SnapshootQueue =
    new ConcurrentQueue<QuoteRt>();

        public static readonly ConcurrentDictionary<string, DateTime> SnapshootUpdateTime =
            new ConcurrentDictionary<string, DateTime>();

        public static readonly ConcurrentDictionary<string, long> SnapshootUpdateDelay =
            new ConcurrentDictionary<string, long>();

        public ConnectPool ConnectPool;

        public List<Stkinfo> StockList;

        public static readonly ConcurrentDictionary<TdxConnection, Task> Tasks =
            new ConcurrentDictionary<TdxConnection, Task>();

        public static readonly ConcurrentDictionary<TdxConnection, HqLooper> Loopers =
            new ConcurrentDictionary<TdxConnection, HqLooper>();

        public Task QueueTask;
        public QueueLooper QueueLooper;

        public Queue<HqLooper> ErrorHqLoopers = new Queue<HqLooper>();
        private readonly object _errorQueueLock = new object();

        public static void Main(string[] args)
        {
            InitIpList();
            var sanpshootService = new SanpshootService();
            Task.Run(() => QuoteToCallback());
            //TaskConfig.RunTask();
            Console.ReadKey();

        }
        private static void QuoteToCallback()
        {
            //while(true)
            //{
            //    Stopwatch watch = new Stopwatch(); watch.Start();
            //    Thread.Sleep(200);
            //    int ic = SanpshootService.SnapshootQueue.Count;
            //    ConcurrentDictionary<string, QuoteRt> Snapshoot = new ConcurrentDictionary<string, QuoteRt>(); SnapshootQueue.
            //    for (int i = 0; i < ic; i++)
            //    {
            //        QuoteRt dd;
            //        var xx = SanpshootService.SnapshootQueue.TryDequeue(out dd);
            //        Snapshoot.AddOrUpdate(dd.StkCode, dd, (k, v) => dd);
            //    }
            //    RedisUtils.GetInstance().HqRedis.GetDatabase().Publish("all", JsonConvert.SerializeObject(Snapshoot));
            //    watch.Stop();
            //    Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString()+":cut:" + ic+"耗时:"+ watch.ElapsedMilliseconds);
            //}
            ConcurrentDictionary<string, QuoteRt> Snapshoot = new ConcurrentDictionary<string, QuoteRt>(); 
            while (true)
            {
                Stopwatch watch = new Stopwatch(); watch.Start();
                if (!SanpshootService.SnapshootQueue.IsEmpty)
                {
                    //从队列中取出  
                    QuoteRt a;
                    if (SanpshootService.SnapshootQueue.TryDequeue(out a))
                    {
                        Snapshoot.AddOrUpdate(a.StkCode, a, (k, v) => a);                       
                    }
                }
                else
                {
                    int i = Snapshoot.Count;
                    if (Snapshoot.Count > 0)
                    {
                        RedisUtils.GetInstance().HqRedis.GetDatabase().Publish("UpdateQuote", JsonConvert.SerializeObject(Snapshoot));
                        Snapshoot.Clear();
                    }
                    Thread.Sleep(200);
                    watch.Stop();
                    Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString()+":Num:" +i +"耗时:"+ watch.ElapsedMilliseconds);
                }
            }
        }

        public SanpshootService()
        {
            RedisUtils.GetInstance();
            QueueLooper = QueueLooper.Instance;
            ConnectPool = ConnectPool.Instance;
            StockList = Quote.TdxGetStkcode(ConnectPool.GetFasterConncetion());

            EmailSenter.SendEmail("行情服务开启", "行情服务开启");

            //初始化队列服务
            QueueTask = Task.Factory.StartNew(() => { QueueLooper.Run(); });

            if (ConnectPool.GetAvailabeConncetion().Count == 0)
            {
                _log.Error("tdxConnections.Count == 0 连接数为0，请检查通达信行情服务是否正常，以及TdxHq.dll是否正常");
                EmailSenter.SendEmail("通达信行情错误", "tdxConnections.Count == 0 连接数为0，请检查通达信行情服务是否正常，以及TdxHq.dll是否正常");
                throw new Exception("connection error");
            }

            //初始化行情服务
            foreach (var pair in Config.IpList)
            {
                List<TdxConnection> tdxConnections = ConnectPool.GetAvailabeConncetion();

                var connectionList = tdxConnections.Where(m => m.Ip.Contains(pair.Key)).ToList();

                if (connectionList.Count == 0)
                {
                    continue;
                }

                var stocksLists = StockList.Partition(StockList.Count / connectionList.Count + 1).ToList();
                for (int i = 0; i < connectionList.Count; i++)
                {
                    var tdxConnection = connectionList[i];
                    var stocksList = stocksLists[i];
                    Task task = Task.Factory.StartNew(() =>
                    {
                        HqLooper hqLooper = null;
                        try
                        {
                            hqLooper = new HqLooper(tdxConnection, stocksList)
                            {
                                Listener = (connec) => throw new Exception(connec.ConnectionName + "error")
                            };
                            Loopers.AddOrUpdate(tdxConnection, hqLooper, (k, v) => hqLooper);
                            hqLooper.Run();
                        }
                        catch (Exception e)
                        {
                            if (null != hqLooper)
                            {
                                lock (_errorQueueLock)
                                {
                                    ErrorHqLoopers.Enqueue(hqLooper);
                                }
                            }
                            _log.Error(e);
                            EmailSenter.SendEmail("通达信行情错误",
                                "HqLooper stop connection name = " + tdxConnection.ConnectionName +
                                "\n" +
                                "ip = " + tdxConnection.Ip + "\n" +
                                "stock list = " +
                                StockList.Select(m => m.Stkname + ":" + m.Stkcode).ToList().List2String() + "\n" +
                                "exception = " + e);
                        }
                    });

                    Tasks.AddOrUpdate(tdxConnection, task, (k, v) => task);
                }
            }


            //初始化行情维护服务
            Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    if (0 != ErrorHqLoopers.Count)
                    {
                        HqLooper hqLooper = null;

                        lock (_errorQueueLock)
                        {
                            hqLooper = ErrorHqLoopers.Dequeue();
                        }

                        _log.Info("restart tdxconnection HqLooper connection name = " +
                                  hqLooper.TdxConnection.ConnectionName +
                                  "\n" +
                                  "ip = " + hqLooper.TdxConnection.Ip + "\n" +
                                  "stock list = " + StockList.List2String());

                        EmailSenter.SendEmail("行情链接重连",
                            "HqLooper stop connection name = " + hqLooper.TdxConnection.ConnectionName +
                            "\n" +
                            "ip = " + hqLooper.TdxConnection.Ip + "\n" +
                            "stock list = " + StockList.Select(m => m.Stkname + ":" + m.Stkcode).ToList()
                                .List2String());

                        try
                        {
                            bool connected = hqLooper.TdxConnection.Connect();
                            if (!connected)
                            {
                                lock (_errorQueueLock)
                                {
                                    ErrorHqLoopers.Enqueue(hqLooper);
                                }
                                _log.Error("restart error HqLooper connection name = " +
                                           hqLooper.TdxConnection.ConnectionName +
                                           "\n" +
                                           "ip = " + hqLooper.TdxConnection.Ip + "\n" +
                                           "stock list = " +
                                           StockList.Select(m => m.Stkname + ":" + m.Stkcode).ToList().List2String());
                            }
                        }
                        catch (Exception e)
                        {
                            _log.Error(e);
                            EmailSenter.SendEmail("行情链接重连错误",
                                "HqLooper restart error\nconnection name = " + hqLooper.TdxConnection.ConnectionName +
                                "\n" +
                                "ip = " + hqLooper.TdxConnection.Ip + "\n" +
                                "stock list = " + StockList.Select(m => m.Stkname + ":" + m.Stkcode).ToList()
                                    .List2String() + "\n" +
                                "exception = " + e);
                        }
                    }
                    Thread.Sleep(10000);
                }
            });

//            Task.Factory.StartNew(() =>
//            {
//                while (true)
//                {
//                    _log.Debug("002926 delay = " + SnapshootUpdateDelay.GetOrAdd("002926", 0) / 10000 + "ms");
//                    _log.Debug("300113 delay = " + SnapshootUpdateDelay.GetOrAdd("300113", 0) / 10000 + "ms");
//                    Thread.Sleep(2000);
//                }
//            });
        }

        private static void InitIpList()
        {
            var json = File.ReadAllText(@".\ip.config");
            var ipDic = JsonConvert.DeserializeObject<Dictionary<string, int>>(json);

            foreach (var keyValuePair in ipDic)
            {
                if (!Config.IpList.ContainsKey(keyValuePair.Key))
                {
                    Config.IpList.Add(keyValuePair.Key, keyValuePair.Value);
                }
            }
        }
    }
}